package com.fulihui.wgj.consumer.listener;

import com.alibaba.dubbo.config.annotation.Reference;
import com.fulihui.common.constants.TopicConstants;
import com.fulihui.common.enums.SendLogStatusEnum;
import com.fulihui.common.result.rpc.BaseResult;
import com.fulihui.common.util.MassMsgIdUtil;
import com.fulihui.weixinsdk.result.mass.MassSendResult;
import com.fulihui.wgj.consumer.service.WxMsgService;
import com.fulihui.wgj.facade.msg.MassMsg;
import com.fulihui.wgj.facade.request.mass.MassLogSendUpdateRequest;
import com.fulihui.wgj.facade.service.mass.WxMassLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author: LiuJie
 * @version: 2018/8/2 10:25
 * @description:
 */
@Component
@Slf4j
public class WxMassMsgListener extends AbstractListener<MassMsg> {
    @Autowired
    private WxMsgService wxMsgService;
    @Autowired
    private MassMsgIdUtil massMsgIdUtil;
    @Reference(version = "1.0.0")
    private WxMassLogService wxMassLogService;

    /**
     * kafka监听方法抽象，规范实现
     *
     * @param record
     */
    @KafkaListener(topics = TopicConstants.MASS_MSG)
    @Override
    void listener(ConsumerRecord<String, String> record) {
        MassMsg msg = super.unpack(record.value(), MassMsg.class);
        execute(msg);
    }

    /**
     * 执行具体业务
     *
     * @param msg 具体消息通配
     */
    @Override
    void execute(MassMsg msg) {
        super.execute(msg);
        MassSendResult massSendResult = wxMsgService.sendMassMsg(msg);
        String msg_id = massSendResult.getMsg_id();
        Long recordId = msg.getRecordId();
        // 将record id 和 msgId 写入
        massMsgIdUtil.addMsgId(msg_id, recordId.toString());

        // 更新发送完成
        if (msg.getLast()) {
            MassLogSendUpdateRequest request = new MassLogSendUpdateRequest();
            request.setId(recordId);
            request.setSendEndTime(new Date());
            request.setSendStatus(SendLogStatusEnum.complete.getCode());
            BaseResult baseResult = wxMassLogService.updateMassLogAccumulate(request);
            log.info("图文群发完成，recordId={},size={},mediaId={},appId={}",
                    msg.getRecordId(), msg.getOpenIds().size(), msg.getMediaId(), msg.getAppId());
            if (!baseResult.isSuccess()) {
                log.error("图文日志更新状态失败，recordId={}", recordId);
            }
        }
    }
}
